package com.shujia.spark.kafka

import kafka.common.TopicAndPartition
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{Cell, CellUtil, HColumnDescriptor, HTableDescriptor}
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes

object OffsetManagerUtil {

  /**
    * 将偏移量保存到hbase中
    * 表名： offset_{groupId}   groupId唯一标识一个任务
    * rowkey:  topic
    * 列名：分区编号
    * 列值： 偏移量
    *
    */

  //1、 创建hbase连接
  //创建hadoop配置文件对象
  val configuration: Configuration = new Configuration
  //指定zk地址
  configuration.set("hbase.zookeeper.quorum", "node1:2181")
  //建立连接   和zk的连接
  val connection: HConnection = HConnectionManager.createConnection(configuration)

  val hBaseAdmin = new HBaseAdmin(configuration)


  //创建保存偏移量的表
  def createTable(groupId: String): Unit = {
    val tableName = "offset_" + groupId

    if (!hBaseAdmin.tableExists(tableName)) {
      println("创建保存偏移量的表")
      val tableDescriptor = new HTableDescriptor(tableName)
      val columnDescriptor = new HColumnDescriptor("info")
      tableDescriptor.addFamily(columnDescriptor)
      //创建表
      hBaseAdmin.createTable(tableDescriptor)
    }
  }

  //更新偏移量
  def updateOffSet(groupId: String, topic: String, partition: Int, offset: Long): Unit = {
    val tableName = "offset_" + groupId
    val table = connection.getTable(tableName)
    //以topic作为rowkey
    val put = new Put(topic.getBytes())
    //增加一列
    put.add("info".getBytes(), Bytes.toBytes(partition), Bytes.toBytes(offset))
    table.put(put)
  }

  //获取偏移量
  def getOffSet(groupId: String): Map[TopicAndPartition, Long] = {

    var fromOffsets: Map[TopicAndPartition, Long] = null

    val tableName = "offset_" + groupId
    val table = connection.getTable(tableName)
    val scan = new Scan()
    val resultSet = table.getScanner(scan)

    //遍历返回的结果
    var result = resultSet.next()
    while (result != null) {
      //导入一个隐式转换    java集合就可以和scala集合进行相互转换了
      import scala.collection.JavaConverters._
      //获取偏移量
      val cells = result.listCells().asScala

      //将数据处理成指定格式
      fromOffsets = cells.map(cell => {
        val topic = Bytes.toString(CellUtil.cloneRow(cell))
        val partition = Bytes.toInt(CellUtil.cloneQualifier(cell))
        val offset = Bytes.toLong(CellUtil.cloneValue(cell))
        (TopicAndPartition(topic, partition), offset)
      }).toMap

      result = resultSet.next()
    }


    //返回偏移量
    fromOffsets
  }

}
